package org.mozhu.simple.rpc.server;

import org.mozhu.simple.rpc.utils.ThreadUtils;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

public class Acceptor extends AbstractServerThread {

    public static final int USE_DEFAULT_BUFFER_SIZE = -1;

    private EndPoint endPoint;
    private Integer sendBufferSize;
    private Integer recvBufferSize;
    private List<Processor> processors;
    private Selector nioSelector;
    private ServerSocketChannel serverChannel;

    public Acceptor(EndPoint endPoint, Integer sendBufferSize, Integer recvBufferSize, List<Processor> processors) {
        this.endPoint = endPoint;
        this.sendBufferSize = sendBufferSize;
        this.recvBufferSize = recvBufferSize;
        this.processors = processors;

        try {
            nioSelector = Selector.open();
        } catch (IOException e) {
            e.printStackTrace();
        }

        serverChannel = openServerSocket(endPoint.getHost(), endPoint.getPort());
        initProcessors();
    }

    private ServerSocketChannel openServerSocket(String host, int port) {
        InetSocketAddress socketAddress;
        if (host == null || host.trim().isEmpty()) {
            socketAddress = new InetSocketAddress(port);
        } else {
            socketAddress = new InetSocketAddress(host, port);
        }

        ServerSocketChannel serverChannel = null;
        try {
            serverChannel = ServerSocketChannel.open();
            serverChannel.configureBlocking(false);
            if (recvBufferSize != USE_DEFAULT_BUFFER_SIZE) {
                serverChannel.socket().setReceiveBufferSize(recvBufferSize);
            }
            serverChannel.socket().bind(socketAddress);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return serverChannel;
    }

    private void initProcessors() {
        for (Processor processor : this.processors) {
            ThreadUtils.newThread(String.format("rpc-network-thread-%d", processor.getId()),
                    processor, false).start();
        }
    }

    @Override
    public void wakeup() {

    }

    @Override
    public void run() {
        try {
            serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT);
        } catch (ClosedChannelException e) {
            e.printStackTrace();
        }

        startupComplete();

        try {
            int currentProcessor = 0;
            while (isRunning()) {
                try {
                    int ready = nioSelector.select(500);
                    if (ready > 0) {
                        Set<SelectionKey> selectionKeys = nioSelector.selectedKeys();
                        Iterator<SelectionKey> iterator = selectionKeys.iterator();
                        while (iterator.hasNext() && isRunning()) {
                            try {
                                SelectionKey selectionKey = iterator.next();
                                iterator.remove();
                                if (selectionKey.isAcceptable()) {
                                    accept(selectionKey, processors.get(currentProcessor));
                                }

                                // round robin to the next processor thread
                                currentProcessor = (currentProcessor + 1) % processors.size();
                            } catch (Throwable t) {
                                t.printStackTrace();
                            }
                        }
                    }
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        } finally {
            shutdownComplete();
        }

    }

    private void accept(SelectionKey selectionKey, Processor processor) {
        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
        try {
            SocketChannel socketChannel = serverSocketChannel.accept();
            socketChannel.configureBlocking(false);
            socketChannel.socket().setTcpNoDelay(true);
            socketChannel.socket().setKeepAlive(true);
            if (sendBufferSize != USE_DEFAULT_BUFFER_SIZE) {
                socketChannel.socket().setSendBufferSize(sendBufferSize);
            }
            processor.accept(socketChannel);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
